package com.shujia.window

import java.lang

import com.alibaba.fastjson.{JSON, JSONObject}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

object Demo4WindowProcess {
  def main(args: Array[String]): Unit = {

    /**
      * 实时统计道路的平均车速，统计最近1分钟额车辆，每隔5秒统计一次
      *
      */
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    val linesDS: DataStream[String] = env.socketTextStream("master", 8888)

    //取出道路编号和车速
    val roadAndSpeedDS: DataStream[(Long, Double)] = linesDS.map(line => {
      val carJson: JSONObject = JSON.parseObject(line)
      val roadId: Long = carJson.getLong("road_id")
      val speed: Double = carJson.getDouble("speed")
      (roadId, speed)
    })

    val windowDS: WindowedStream[(Long, Double), Long, TimeWindow] = roadAndSpeedDS
      .keyBy(_._1)
      //滑动的处理时间窗口
      .window(SlidingProcessingTimeWindows.of(Time.minutes(1), Time.seconds(5)))

    /**
      * process: flink底层api
      * 可以在process中操作flink的事件，时间，状态
      *
      */

    val avgSpeedDS: DataStream[(Long, Double)] = windowDS.process(new MyWindowProcessFunction)

    avgSpeedDS.print()


    env.execute()

  }
}

class MyWindowProcessFunction extends ProcessWindowFunction[(Long, Double), (Long, Double), Long, TimeWindow] {

  /**
    * process: 每一个窗口会调用一次，会将整个窗口内的数据传递给这个函数
    *
    * @param key      ： keyBy的key
    * @param context  : 上下文对象
    * @param elements ： 窗口内的数据
    * @param out      ： 用于将数据发生到下游
    */
  override def process(key: Long,
                       context: Context,
                       elements: Iterable[(Long, Double)],
                       out: Collector[(Long, Double)]): Unit = {


    //记录总的车速
    var sumSpeed = 0.0
    //记录车的数量
    var count = 0

    //循环计算总的车速和车的数量
    for ((road, speed) <- elements) {
      sumSpeed += speed
      count += 1
    }

    //计算平均车速
    val avgSpeed: Double = sumSpeed / count

    //将数据发生到下游
    out.collect((key, avgSpeed))
  }
}
